<?php
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;
use PhpAmqpLib\Message\AMQPMessage;

include(__DIR__.'/config.php');

$exchange_name = 'someExchange';
$queue_name    = 'someQueue';

//创建连接实力
$connection = new AMQPStreamConnection(HOST,PORT,USER,PASS,VHOST);

//创建通道
$channel = $connection->channel();

$args = new AMQPTable(['x-max-length' => 10]);


// 开启confirm模式，confirm模式下，投递消息后，RabbitMQ会异步返回是否投递成功，confirm模式不可以和事务模式同时存在
$channel->confirm_select();

// 推送消息到RabbitMQ成功的异步回调，如果消息推送成功，想做什么业务处理写在这里
$channel->set_ack_handler(function (AMQPMessage $message){
    echo "Message acked with content " . $message->body . PHP_EOL;
});

// 推送消息到RabbitMQ失败的异步回调，如果消息推送失败，想做什么业务处理写在这里
$channel->set_nack_handler(function (AMQPMessage $message){
    echo "Message nacked with content " . $message->body . PHP_EOL;
});

//声明交换机，将第四个参数设置为true，表示将交换机持久化
$channel->exchange_declare($exchange_name,AMQP_EX_TYPE_FANOUT,false,false,true);

//声明队列名称，将第三个参数设置为true，表示将队列持久化
$channel->queue_declare($queue_name,false, true, false, false ,false ,$args);

//队列与交换机绑定，因为是扇形交换机，直接与队列绑定，无需与路由键绑定
$channel->queue_bind($queue_name,$exchange_name);

$i = 1;

while ($i <= 100){
    //消息必须是string或者int类型
    $content = '这是消息内容' . $i++;

    //消息类，设置delivery_mode为DELIVERY_MODE_PERSISTENT，表示将消息持久化
    $msg = new AMQPMessage($content,[
        'content_type' => 'text/plain',
        'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
    ]);

    //消息推送
    $channel->basic_publish($msg,$exchange_name);
}

//监听写入
$channel->wait_for_pending_acks();

$channel->close();

$connection->close();


